#pragma once

#include "../common/Log.hpp"
#include "../common/message.pb.h"
#include "Consumer.hpp"
#include <string>
#include <unordered_map>
#include <memory>
#include <vector>
#include <mutex>
#include <functional>

namespace micromq
{
    class ConsumerCenter
    {
    public:
        using ptr = std::shared_ptr<ConsumerCenter>;
    public:
        ConsumerCenter(const std::string& msgQueueName): _msgQueueName(msgQueueName), _rotatingSeq(0)
        {}

        Consumer::ptr addConsumer(const std::string& tag, const std::string& msgQueueName, 
            bool autoAck, const ConsumerCallback& cb)
        {
            std::unique_lock<std::mutex> lock(_mtx);
            for(auto& a : _consumers)
            {
                if(a->_tag == tag)
                {
                    logInfo("consumer is already in consumer center...");
                    return nullptr;
                }
            }
            Consumer::ptr consumer = std::make_shared<Consumer>(tag, msgQueueName, autoAck, cb);
            _consumers.push_back(consumer);
            return consumer;
        }

        void removeConsumer(const std::string& tag)
        {
            std::unique_lock<std::mutex> lock(_mtx);
            for(auto it = _consumers.begin(); it != _consumers.end(); ++it)
            {
                if((*it)->_tag == tag)
                {
                    _consumers.erase(it);
                    return;
                }
            }
        }

        Consumer::ptr getConsumer()
        {
            std::unique_lock<std::mutex> lock(_mtx);
            if(_consumers.size() == 0) return nullptr;
            uint64_t index = _rotatingSeq++ % _consumers.size();
            return _consumers[index];
        }

        bool isEmpty()
        {
            std::unique_lock<std::mutex> lock(_mtx);
            return _consumers.size() == 0;
        }

        bool isExists(const std::string& tag)
        {
            std::unique_lock<std::mutex> lock(_mtx);
            for(auto& a : _consumers)
            {
                if(a->_tag == tag) return true;
            }
            return false;
        }

        void clear()
        {
            std::unique_lock<std::mutex> lock(_mtx);
            _rotatingSeq = 0;
            _consumers.clear();
        }

        ~ConsumerCenter()
        {}
    private:
        std::mutex _mtx;
        std::string _msgQueueName;
        uint64_t _rotatingSeq; //轮转序号，负载均衡
        std::vector<Consumer::ptr> _consumers; 
    };
}